package com.acmedcare.framework.exchange.center.executer.listener;

import com.acmedcare.framework.exchange.center.entity.bean.Task;
import com.acmedcare.framework.exchange.center.entity.dto.ActionCommand;
import com.acmedcare.framework.exchange.center.entity.dto.CommandEvent;
import com.acmedcare.framework.exchange.center.entity.dto.TaskCommandEvent;
import com.acmedcare.framework.exchange.center.entity.dto.TaskStatusEnum;
import com.acmedcare.framework.exchange.center.executer.ExecutorConfig;
import com.acmedcare.framework.exchange.center.executer.TaskRunnerManager;
import com.acmedcare.framework.exchange.center.executer.job.NormalExecuteJob;
import com.acmedcare.framework.exchange.center.executer.job.PluginJob;
import com.acmedcare.framework.exchange.center.executer.job.SparkExecuteJob;
import com.acmedcare.framework.starter.bus.AcmedcareEventListener;
import com.acmedcare.framework.starter.bus.AcmedcareEventPublisher;
import lombok.extern.slf4j.Slf4j;
import org.quartz.CronScheduleBuilder;
import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class CommandListener extends AcmedcareEventListener<CommandEvent> {

  final RabbitTemplate rabbitTemplate;

  public CommandListener(AcmedcareEventPublisher publisher,
      RabbitTemplate rabbitTemplate) {
    super(publisher);
    this.rabbitTemplate = rabbitTemplate;
  }

  @Override
  public void onAcmedcareEvent(CommandEvent commandEvent) {

    final TaskRunnerManager manager = new TaskRunnerManager();
    Task task = commandEvent.getTask();
    JobKey key = JobKey.jobKey(task.getName(), ExecutorConfig.TASKGROUP);
    Scheduler scheduler = ExecutorConfig.getInstance().getScheduler();

    TaskStatusEnum statusEnum = TaskStatusEnum.STOPED;
    manager.addTask(task);

    try {

      switch (commandEvent.getCommandType()) {

        case STOP:
          if (scheduler.checkExists(key)) {

            log.info("delete job success,task name:" + task.getName());
            scheduler.deleteJob(key);
          }

          statusEnum = TaskStatusEnum.STOPED;
          break;

        case START:
        case RESTART:
          if (scheduler.checkExists(key)) {

            scheduler.deleteJob(key);
            log.info("resume job success,task name:" + task.getName());
          }

          task.setStatus(TaskStatusEnum.STARTED);
          createJob(task, scheduler, key);
          log.info("create and restart job success,task name:" + task.getName());

          statusEnum = TaskStatusEnum.STARTED;
          break;

        case PAUSE:
          if (scheduler.checkExists(key)) {

            scheduler.pauseJob(key);
            log.info("pause job success,task name:" + task.getName());
          } else {

            log.warn("pause job is not exists,task name:" + task.getName());
          }

          statusEnum = TaskStatusEnum.PAUSE;
          break;
      }
    } catch (Exception e) {

      log.error("执行采集任务异常:" + e.getMessage(), e);
    }

    task.setStatus(statusEnum);
    TaskCommandEvent taskCommandEvent = new TaskCommandEvent();
    taskCommandEvent.setActionCommand(ActionCommand.Update);
    taskCommandEvent.setData(task);
    publisher.publishEvent(ExecutorConfig.EXCHANGENAME, taskCommandEvent);
  }

  private void createJob(Task task, Scheduler scheduler, JobKey key) {

    Class<? extends Job> jobClass;

    switch (task.getSchedulerType()) {
      case NORMAL:
        jobClass = NormalExecuteJob.class;
        break;
      case AGENT:
        jobClass = PluginJob.class;
        // 如果任务是采集器器任务，则先行创建交换器
        RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());
        admin.declareExchange(new FanoutExchange(task.getName()));
        break;
      case SPARK:
      default:
        jobClass = SparkExecuteJob.class;
        break;
    }

    JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(key).build();
    Trigger trigger = TriggerBuilder.newTrigger()
        .withIdentity(task.getName() + "Trigger", ExecutorConfig.getInstance().getGroupName())
        .withSchedule(CronScheduleBuilder.cronSchedule(task.getExecuteCron())).build();
    jobDetail.getJobDataMap().put(ExecutorConfig.TASKINFO, task);
    try {
      scheduler.scheduleJob(jobDetail, trigger);
    } catch (SchedulerException e) {
      log.error("create job fail", e);
    }
  }
}
